package com.disruptor.disruptorStudy.demo03;

import java.nio.IntBuffer;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import com.disruptor.disruptorStudy.demo01.Trade;
import com.disruptor.disruptorStudy.demo01.TradeHandler01;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

public class Test02 {

    public static void main(String[] args) throws Exception {
        AtomicInteger count = new AtomicInteger(0);
        //创建5个消费者
        TradeHandler01[] handlers = new TradeHandler01[5];
        handlers[0] = new TradeHandler01("侯征");
        handlers[1] = new TradeHandler01("姚振");
        handlers[2] = new TradeHandler01("李世彪");
        handlers[3] = new TradeHandler01("熊正航");
        handlers[4] = new TradeHandler01("崔杰");
        //不重复消费数据
        Disruptor<Trade> disruptor =
                new Disruptor<Trade>(() -> {
                    return new Trade();
                },
                        1024, Executors.newFixedThreadPool(4),
                        ProducerType.MULTI, new YieldingWaitStrategy());
        // 连接消费事件方法,多消费者
        //handleEventsWithWorkerPool: 分组消费,即不重复消费数据,使用WorkHandler的实现类消费者
        //disruptor.handleEventsWithWorkerPool(handlers);
        // handleEventsWith: 广播消费: 即重复消费数据 使用EventHandler接口的实现类消费者
        disruptor.handleEventsWith(handlers);
        disruptor.start();
        //生产者生产数据
        final CountDownLatch latch = new CountDownLatch(1);
        for (int i = 0; i < 10; i++) {  //创建一百个生产者生产数据
            Produces p = new Produces(disruptor.getRingBuffer());
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        latch.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //使用IntBuffer,每个生产者生产十个数据
                    IntBuffer byteBuffer = IntBuffer.allocate(8);
                    for (int j = 0; j < 10; j++) {
                        byteBuffer.put(new Random().nextInt(100));
                        byteBuffer.flip();
                        count.addAndGet(1);//计算生产数据总数
                        p.onData(byteBuffer);
                    }
                }
            }).start();
        }
        //100个生产者全部创建完成后再一起同时生产数据(即通知await)
        latch.countDown();
        Thread.sleep(5000);//等待生产并消费数据
        disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
        System.out.println("第一个消费者消费总数:" + handlers[0].getCount());
        System.out.println("第二个消费者消费总数:" + handlers[1].getCount());
        System.out.println("第三个消费者消费总数:" + handlers[2].getCount());
        System.out.println("第四个消费者消费总数:" + handlers[3].getCount());
        System.out.println("第五个消费者消费总数:" + handlers[4].getCount());
        int total = 0;
        for (int i = 0; i < handlers.length; i++) {
            total += handlers[i].getCount();
        }
        System.out.println("总共生产了:" + count.get() + "个数据");
        System.out.println("总共消费:" + total);//每个消费者都会消费一遍所有的数据
    }

}
